package org.jgroups.tests;

import org.jgroups.*;
import org.jgroups.logging.Log;
import org.jgroups.logging.LogFactory;
import org.jgroups.protocols.FragHeader3;
import org.jgroups.protocols.FRAG3;
//import org.apache.commons.logging.Log;
//import org.apache.commons.logging.LogFactory;

import java.io.File;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.List;
import java.util.ArrayList;

import static org.jgroups.tests.TestSender.*;

/**
 * Created by IntelliJ IDEA.
 * User: thsieh
 * Date: Mar 17, 2009
 * Time: 11:41:45 AM
 * To change this template use File | Settings | File Templates.
 */
public class TestReceiver extends ReceiverAdapter implements ChannelListener {
    private static Log logger = LogFactory.getLog(TestReceiver.class);


    private int count ;
    private AtomicInteger no ;
    private FRAG3 frag3;
    private List<Long> begin = new ArrayList<Long>();
    private List<Long> current;
    private String culster ;
    private JChannel ch;

    public TestReceiver(int count) {
        this.count = count ;
        this.no = new AtomicInteger(0);
        List<String> list = runShell(new String[] { "sh","-c", "netstat -s | grep -A4 Udp" }) ;
        for ( String each : list) {
            String[] strs = each.split("\\s");
            begin.add( Long.valueOf( strs[0]) );
        }
        current = begin;
        logger.info( output(begin));
    }

    private void logMsg(Message msg) {
        FragHeader3 hdr = (FragHeader3) msg.getHeader(FRAG3.name );
        if ( hdr != null && hdr.type == FragHeader3.MSG_STREAM) {
            long len = (long) ( (hdr.num_frags - 1 ) * frag3.getFragSize()) +
                    (long) (hdr.remain == frag3.getFragSize() ? frag3.getFragSize() : hdr.remain) ;
            deleteFile( msg);
            logger.info("Receive message source=" +msg.getSrc().toString()+" length="+ len  +" destination="+ msg.getDest()
                    +" "+ hdr.toString() ) ;
        }
        else logger.info("Receive message source=" +msg.getSrc().toString()+" length="+ msg.getLength()
                +" destination="+ msg.getDest()+" "+ ( hdr == null ? "null" : hdr.toString()));

    }

    public void receive(Message msg) {
        if ( msg == null ) logger.warn("Mesasge is null ");
        else logMsg( msg);
        int i = no.getAndIncrement();
        if ( i % count == (count -1 ) ) {
            List<Long> now =  new ArrayList<Long>();
            logger.info("FRAG3 " + frag3.toString() );
            List<String> list = runShell(new String[] { "sh","-c", "netstat -s | grep -A4 Udp" }) ;
            for ( String each : list) {
                String[] strs = each.split("\\s");
                now.add( Long.valueOf( strs[0]) );
            }
            // check array size
            if ( now.size() >= 2 && current.size() >= 2) {
                long dif = now.get(2) - current.get(2);
                if ( dif == 0 ) logger.info(i+" messages no udp error");
                else logger.info(i+" messages extra received udp error " + dif);
            }
            current = now;
        }
    }

    private void deleteFile(Message msg) {
        try {
            String payload = new String(msg.getRawBuffer());
            // split with space, 1st - inter filename , 2nd original filename
            String[] files = payload.split(" ");
            if ( files.length == 2 ) {
                logger.info("internal "+ files[0]+" original "+ files[1]+" delete "+ files[0]);
                File f = new File( files[0]);
                if ( ! f.delete() ) logger.error("delete " + files[0]+" fail ");
            }
            else {
                logger.error("payload error "+ payload );
            }
        } catch ( Exception ex) {
            logger.error( ex.getMessage() );
        }
    }

    @Override
	public void viewAccepted(View view) {
			logger.info("ViewAccepted size=>"+ view.size()+"  "+view.printDetails() );
	}

    public static void help() {
          System.out.println("java stress.org.jgroups.test.TestSender -udp ./udp.xml -cluster test-1 -count 100 -size 10");
    }

    public void channelConnected(Channel channel) {
        logger.info("channelConnected " + channel.getClusterName() );
    }

    public void channelDisconnected(Channel channel){
        logger.info("channelDisconnected " + channel.getClusterName() );
    }

    public void channelClosed(Channel channel) {
        logger.info("channelClosed " + channel.getClusterName() );
    }
    public void channelShunned() {
        logger.info("channelShunned "  );

    }

    public void channelReconnected(Address addr) {
        logger.info("channelReconnected " + addr.toString() );

    }


    public static void main(String[] args) throws Exception {
        String[] defaults = { "./udp.xml","test-1" , "100" , "20", "2000" };

        for ( int i=0; i < args.length ; ) {
            if ( args[i].equals("-udp")) defaults[0] = args[++i] ;
            else if ( args[i].equals("-cluster")) defaults[1] = args[++i] ;
            else if ( args[i].equals("-count")) defaults[2] = args[++i] ;
            else if ( args[i].equals("-size")) defaults[3] = args[++i] ;
            else if ( args[i].equals("-times")) defaults[4] = args[++i] ;
            else if ( args[i].equals("-help")) help() ;
            else {
               help();
            }
            i ++;
        }
        int c =  Integer.valueOf( defaults[2]);
        System.setProperty("java.net.preferIPv4Stack", "true");
        logger.info("java.net.preferIPv4Stack ="+ System.getProperty("java.net.preferIPv4Stack"));
        TestReceiver tr = new TestReceiver(c);
        tr.ch = new JChannel(new File(defaults[0]));
        tr.ch.setReceiver(tr);
        tr.culster = defaults[1];
        tr.ch.connect(tr.culster);
        tr.frag3 = (FRAG3) tr.ch.getProtocolStack().getTopProtocol();
        logger.info("Address "+tr.ch.getAddress().toString());
//        tr.ch.handleExit(new Event(Event.EXIT));
//        System.out.println("sleep 2 seconds to reconnect");
//        Thread.sleep(2000);
//        tr.ch.open();
//        tr.ch.connect(tr.culster);
    }

}
